package src

import (
	"bytes"
	"encoding/json"
	"io"
	"io/ioutil"
	"net/http"
	"sync"
	"time"
	"webmsg/config"

	"github.com/gorilla/websocket"
	uuid "github.com/satori/go.uuid"
	"github.com/sirupsen/logrus"
)

const (
	// Time allowed to write a message to the peer.
	writeWait = 10 * time.Second

	// Time allowed to read the next pong message from the peer.
	pongWait = 60 * time.Second

	// Send pings to peer with this period. Must be less than pongWait.
	pingPeriod = (pongWait * 9) / 10

	// Maximum message size allowed from peer.
	maxMessageSize = 512
)

var (
	newline = []byte{'\n'}
	sep = []byte(string(rune(1)))
	space   = []byte{' '}
	onceCM  sync.Once
)

var upgrader = websocket.Upgrader{
	ReadBufferSize:  1024,
	WriteBufferSize: 1024,
	CheckOrigin: func(r *http.Request) bool {
		return true
	},
}

type ClientRoomPair struct {
	client *Client
	roomId string
}

type ReportMessage struct {
	Event    string `json:"event"`
	ClientId string `json:"client_id"`
	UserId   string `json:"user_id"`
	Attach   string `json:"attach"`
}

func NewClientRoomPair(c *Client, ri string) ClientRoomPair {
	return ClientRoomPair{
		client: c,
		roomId: ri,
	}
}

// readPump pumps messages from the websocket connection to the hub.
//
// The application runs readPump in a per-connection goroutine. The application
// ensures that there is at most one reader on a connection by executing all
// reads from this goroutine.
func (c *Client) readPump() error {
	defer func() {
		c.hub.disconnect <- c
		_ = c.conn.Close()
		c.readPumpStopped <- true
	}()
	c.conn.SetReadLimit(maxMessageSize)
	err := c.conn.SetReadDeadline(time.Now().Add(pongWait))
	if err != nil {
		return err
	}
	c.conn.SetPongHandler(
		func(string) error {
			err = c.conn.SetReadDeadline(time.Now().Add(pongWait))
			if err != nil {
				return err
			}
			return nil
		})
	for {
		_, message, err := c.conn.ReadMessage()
		if err != nil {
			if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway, websocket.CloseAbnormalClosure) {
				logrus.WithFields(logrus.Fields{"location": "server", "method": "readPump"}).Warn("Unexpected close")
			}
			return err
		}
		message = bytes.TrimSpace(bytes.Replace(message, newline, space, -1))

		c.handleMessage(message)
	}
}

// writePump pumps messages from the hub to the websocket connection.
//
// A goroutine running writePump is started for each connection. The
// application ensures that there is at most one writer to a connection by
// executing all writes from this goroutine.
func (c *Client) writePump() error {
	ticker := time.NewTicker(pingPeriod)
	defer func() {
		ticker.Stop()
		_ = c.conn.Close()
		c.writePumpStopped <- true
	}()
	var (
		err error
		w   io.WriteCloser
	)
	for {
		select {
		// 消息处理
		case message, ok := <-c.send:
			err = c.conn.SetWriteDeadline(time.Now().Add(writeWait))
			if err != nil {
				return err
			}
			if !ok {
				// The hub closed the channel.
				err = c.conn.WriteMessage(websocket.CloseMessage, []byte{})
				if err != nil {
					return err
				}
				return nil
			}

			w, err = c.conn.NextWriter(websocket.TextMessage)
			if err != nil {
				return err
			}
			_, err = w.Write(message)
			if err != nil {
				return err
			}

			// Add queued chat messages to the current websocket message.
			n := len(c.send)
			for i := 0; i < n; i++ {
				_, err = w.Write(sep)
				if err != nil {
					return err
				}
				_, err = w.Write(<-c.send)
				if err != nil {
					return err
				}
			}

			err = w.Close()
			if err != nil {
				return err
			}
		// 心跳
		case <-ticker.C:
			err = c.conn.SetWriteDeadline(time.Now().Add(writeWait))
			if err != nil {
				return err
			}
			if err = c.conn.WriteMessage(websocket.PingMessage, nil); err != nil {
				return err
			}
		}
	}
}

// 处理消息发送
func (c *Client) handleMessage(message []byte) {
	var payload Message
	err := json.Unmarshal(message, &payload)
	if err != nil {
		// 不是json字符串，则是普通字符串，普通字符串目前仅允许ping
		if string(message) != "ping" {
			return
		}

		c.send <- []byte{'p', 'o', 'n', 'g'}
		return
	}
	// 没有内容，什么都不做
	if payload.Content == "" {
		return
	}
	switch payload.To {
	// 广播，直接发
	case ToGlobal:
		c.hub.broadcast <- []byte(payload.Content)
	// 系统交互，只回复连接端
	case ToSystem:
		switch payload.Type {
		case TypeRegister:
			// 用户身份标识也可以认为是用户自己单独一个房间（废弃）
			// 这里单独开启一个channel处理用户注册是为了方便追踪程序活动，便于监控调试
			c.hub.register <- NewClientRoomPair(c, payload.Content)
		case TypeJoin:
			// 非协程安全，需要用channel来解决冲突，下同
			c.hub.join <- NewClientRoomPair(c, payload.Content)
			c.send <- []byte{'o', 'k'}
		case TypeLeave:
			c.hub.leave <- NewClientRoomPair(c, payload.Content)
			c.send <- []byte{'o', 'k'}
		}
	// 发向某个频道/某个人/某组用户
	default:
		c.hub.Push(payload.To, payload.Type, []byte(payload.Content))
	}
}

// ServeWs serveWs handles websocket requests from the peer.
func ServeWs(hub *Hub, w http.ResponseWriter, r *http.Request) {
	// 检查是否在连接时立即进行注册
	err := r.ParseForm()
	if err != nil {
		logrus.WithFields(logrus.Fields{"location": "server", "method": "ServeWs"}).Warn(err)
		return
	}
	uid := r.Form.Get("uid")
	attach := r.Form.Get("attach")
	if c, ok := cm.registeredClients.Load(uid); uid != "" && ok {
		// 连接的同时再注册，并且存在已经使用同一个user_id注册过的客户端。
		// 让之前的客户端连接断开并给其建立新的连接
		logrus.WithFields(logrus.Fields{"location": "server", "method": "ServeWs"}).Info("Will reset connection.")
		resetConnection(w, r, c.(*Client), uid, attach)
		return
	}
	logrus.WithFields(logrus.Fields{"location": "server", "method": "ServeWs"}).Info("Will start new connection.")
	newConnection(hub, w, r, uid, attach)
}

func NewReportMessage(event string, clientId string, userId string, attach string) ReportMessage {
	return ReportMessage{
		Event:    event,
		ClientId: clientId,
		UserId:   userId,
		Attach:   attach,
	}
}

// Report 报告连接活动
func Report(hub *Hub, event string, clientId string, userId string, attach string) {
	defer func() {
		if err := recover(); err != nil {
			logrus.WithFields(logrus.Fields{"method": "Report"}).Warn(err)
		}
	}()

	cfg, err := config.GetConfig()
	if err != nil {
		logrus.WithFields(logrus.Fields{"method": "Report"}).Warn(err)
	}

	if len(cfg.MonitorApi) == 0 && len(cfg.MonitorRoom) == 0 {
		return
	}

	reportMessage, err := json.Marshal(NewReportMessage(event, clientId, userId, attach))
	if err != nil {
		logrus.WithFields(logrus.Fields{"method": "Report"}).Warn(err)
	}

	logrus.WithFields(logrus.Fields{"method": "Report"}).Info(string(reportMessage))

	// 回调监控api
	if len(cfg.MonitorApi) > 0 {
		msg := bytes.NewBuffer(reportMessage)
		http.DefaultClient.Timeout = time.Second * 1
		for _, u := range cfg.MonitorApi {
			resp, err := http.DefaultClient.Post(u, "application/json", msg)
			if err != nil {
				logrus.WithFields(logrus.Fields{"method": "Report"}).Warn(err)
			}
			if resp != nil {
				body, err := ioutil.ReadAll(resp.Body)
				if err != nil {
					logrus.WithFields(logrus.Fields{"method": "Report"}).Warn(err)
				}
				logrus.WithFields(logrus.Fields{"method": "Report"}).Info("Monitor api "+u+" response: ", string(body))

				_ = resp.Body.Close()
			} else {
				logrus.WithFields(logrus.Fields{"method": "Report"}).Warn("Report api response nil")
			}
		}
	}

	// 推送监控频道
	if len(cfg.MonitorRoom) > 0 {
		for _, roomId := range cfg.MonitorRoom {
			hub.roomManager.Push(roomId, reportMessage)
		}
	}
}

// 创建新的连接
func newConnection(hub *Hub, w http.ResponseWriter, r *http.Request, uid string, attach string) {
	var (
		client *Client
		err    error
		conn   *websocket.Conn
	)

	conn, err = upgrader.Upgrade(w, r, nil)
	if err != nil {
		logrus.WithFields(logrus.Fields{"location": "server", "method": "newConnection"}).Warn(err)
		return
	}
	// 不需要注册或者需要注册但uid未经注册
	cid := uuid.NewV4().String()
	client = NewClient(hub, conn, cid, attach)
	if uid != "" {
		client.hub.Register(client, uid)
	}
	client.hub.connect <- client
	startPump(client)
}

// 重置某个客户端的连接，重置了之后，要启动那两个泵
func resetConnection(w http.ResponseWriter, r *http.Request, client *Client, uid string, attach string) {
	var (
		err  error
		conn *websocket.Conn
	)

	if client.conn != nil {
		_ = client.conn.Close()
		// 等待这个客户端彻底被关闭
		<-client.readPumpStopped
		<-client.writePumpStopped
		<-client.closed

		logrus.WithFields(logrus.Fields{"location": "server", "method": "resetConnection"}).Info("Old client has been closed, refresh it.")
	}

	conn, err = upgrader.Upgrade(w, r, nil)
	if err != nil {
		logrus.WithFields(logrus.Fields{"location": "server", "method": "resetConnection"}).Warn(err)
		return
	}
	client.Refresh(conn)
	// 客户端焕然一新，重新注册并加入连接列表
	client.hub.Register(client, uid)
	client.hub.connect <- client
	client.attach = attach
	startPump(client)
}

func startPump(client *Client) {
	// Allow collection of memory referenced by the caller by doing all work in
	// new goroutines.
	go func() {
		defer func() {
			if err := recover(); err != nil {
				logrus.WithFields(logrus.Fields{"location": "server", "method": "ServeWs"}).Warn(err)
			}
		}()
		err := client.writePump()
		if err != nil {
			logrus.WithFields(logrus.Fields{"location": "server", "method": "ServeWs/writePump"}).Warn(err)
		}
		// writePump退出之前，会把连接关闭，这样依赖，readPump就无法读取信息了，也就自然退出了
	}()
	go func() {
		defer func() {
			if err := recover(); err != nil {
				logrus.WithFields(logrus.Fields{"location": "server", "method": "ServeWs"}).Warn(err)
			}
		}()
		err := client.readPump()
		if err != nil {
			logrus.WithFields(logrus.Fields{"location": "server", "method": "ServeWs/readPump"}).Warn(err)
		}
		// readPump退出之前，会把当前这个客户端压倒unregister队列中，进入到unregister队列中的客户端，都会被关掉它的send队列，
		// send队列关闭会引起writePump关闭
	}()
}
